// Copyright (c) 2020-present,  INSPUR Co, Ltd.  All rights reserved.
// This source code is licensed under Apache 2.0 License.

#include "range_del_mem_arena.h"
namespace rocksdb{
    RangeDelMemArena::RangeDelMemArena(const InternalKeyComparator& mkeyCmp, const ImmutableCFOptions& ioptions)
        : IMemoryArena(mkeyCmp, ioptions),
          mkey_cmp_(mkeyCmp),
          moptions_(ioptions){
    }

    bool RangeDelMemArena::Insert(const Slice &key, const Slice &value) {
      size_t key_size = key.size();
      size_t val_size = value.size();
      const size_t encoded_len = VarintLength(key_size) +
                                 key_size + VarintLength(val_size) +
                                 val_size;
      char* buf = new char[encoded_len];
      char* p = EncodeVarint32(buf, key_size);
      memcpy(p, key.data(), key_size);
      p += key_size;
      p = EncodeVarint32(p, val_size);
      memcpy(p, value.data(), val_size);

      auto node = new VersionNode();
      node->SetNext(nullptr);
      node->SetPrev(nullptr);
      bool ok = node->CASSetKey(nullptr, buf);
      assert(ok);

      bool res = range_del_list_.Insert(reinterpret_cast<const char *>(node));
      if(!res){
        delete[] buf;
      }
      return res;
    }

    ArtTree::Iterator *RangeDelMemArena::Seek(const Slice &key) {
      auto iter = range_del_list_.GetIterator();
      iter->HashSeek(key);
      return iter;
    }

    RangeDelMemArena::~RangeDelMemArena() {
      auto del_iter = range_del_list_.GetIterator();
      del_iter->SeekToFirst();
      while(del_iter->Valid()){
        delete[] del_iter->key();
        del_iter->Next();
      }
    }

//    std::shared_ptr<InternalIterator> RangeDelMemArena::NewMemIter(bool use_range_del_table) {
//      return std::make_shared<InternalIterator>(&range_del_list_, moptions_.inplace_update_support);
//    }


    int RangeDelMemArena::RangeDelKeyComparator::operator()(const char *prefix_len_key1, const char *prefix_len_key2) const {
      // Internal keys are encoded as length-prefixed strings.
      Slice k1 = GetLengthPrefixedSlice(prefix_len_key1);
      Slice k2 = GetLengthPrefixedSlice(prefix_len_key2);
      return comparator.CompareKeySeq(k1, k2);
    }

    int RangeDelMemArena::RangeDelKeyComparator::operator()(const char *prefix_len_key,
                                                            const MemTableRep::KeyComparator::DecodedType &key) const {
      // Internal keys are encoded as length-prefixed strings.
      Slice a = GetLengthPrefixedSlice(prefix_len_key);
      return comparator.CompareKeySeq(a, key);
    }

}  //namespace rocksdb